package ink.xiaobaibai.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import ink.xiaobaibai.emqtt.EMqttManager;
import ink.xiaobaibai.emqtt.EMqttService;
import ink.xiaobaibai.emqtt.MyMqttCallback;
import ink.xiaobaibai.entity.DeviceLog;
import ink.xiaobaibai.entity.MonthlyfeeDevice;
import ink.xiaobaibai.entity.OzoneDevice;
import ink.xiaobaibai.service.IDeviceLogService;
import ink.xiaobaibai.service.IJumpService_V2;
import ink.xiaobaibai.service.IMonthlyfeeDeviceService;
import ink.xiaobaibai.service.IOzoneDeviceService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description: 心跳中心2.0版
 * @author: 小白白
 * @create: 2021-06-12
 **/

@Service
@Slf4j
public class JumpService_V2 implements IJumpService_V2, ApplicationRunner {

    private ThreadGroup logThreadGroup = new ThreadGroup("log-thread-group");
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 10, TimeUnit.MINUTES, new ArrayBlockingQueue<>(100), r -> new Thread(this.logThreadGroup, "log-thread-" + this.threadNumber.getAndIncrement()));

    @Autowired
    private IDeviceLogService deviceLogService;

    @Value("${myApp.dev}")
    private String dev;

    private EMqttService onePayClient;

    private EMqttService monthlyClient;

    private EMqttService buyClient;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        String onePayJumpClientId = "onePayJump";
        String monthlyJumpClientId = "monthlyJump";
        String buyJumpClientId = "buyJump";

        if (true) {
            // 废弃
            return;
        }
        if (this.dev.equals("2")) {
            return;
        }
        if (this.dev.equals("1")) {
            onePayJumpClientId = "onePayJump2";
            monthlyJumpClientId = "monthlyJump2";
            buyJumpClientId = "buyJump2";
        }

        String onePayJumpTopicPub = "onePayJumpPub";
        String onePayJumpTopicSub = "onePayJumpSub";
        this.onePayClient = EMqttManager.getClient(onePayJumpTopicSub, onePayJumpTopicPub, onePayJumpClientId, new OnePayCallback());
        this.onePayClient.start();

        String monthlyJumpTopicPub = "monthlyJumpPub";
        String monthlyJumpTopicSub = "monthlyJumpSub";
        this.monthlyClient = EMqttManager.getClient(monthlyJumpTopicSub, monthlyJumpTopicPub, monthlyJumpClientId, new MonthlyCallback());
        this.monthlyClient.start();

        String buyJumpTopicPub = "buyJumpPub";
        String buyJumpTopicSub = "buyJumpSub";
        this.buyClient = EMqttManager.getClient(buyJumpTopicSub, buyJumpTopicPub, buyJumpClientId, new BuyCallback());
        this.buyClient.start();
    }

    @Autowired
    private IOzoneDeviceService ozoneDeviceService;

    @Autowired
    private IMonthlyfeeDeviceService monthlyfeeDeviceService;

    /**
     * 收集近期的心跳信号obj,放入Map
     * obj: code date
     * code->date 判断key是否存在 和 date 的时间,就可以做到心跳与业务解耦
     */

    //心跳间隔(分钟)
    private final Integer jumpInterval = 10;

    private Map<String, Date> onePayJumpMap = new HashMap<>();

    private Map<String, Date> monthlyJumpMap = new HashMap<>();

    private Map<String, Date> buyJumpMap = new HashMap<>();

    @Override
    public void refreshMap() {
        this.refreshMap(this.onePayJumpMap);
        this.refreshMap(this.monthlyJumpMap);
        this.refreshMap(this.buyJumpMap);
    }

    private void receiveOnePayJump(String code) {
        this.onePayJumpMap.put(code, new Date());
        this.saveLog(code);
    }

    private void receiveMonthlyJump(String code) {
        this.monthlyJumpMap.put(code, new Date());
        this.saveLog(code);
    }

    private void receiveBuyJump(String code) {
        this.buyJumpMap.put(code, new Date());
        this.saveLog(code);
    }

    private void saveLog(String code) {
        this.threadPoolExecutor.execute(() -> this.deviceLogService.save(new DeviceLog(code)));
    }

    @Override
    public boolean checkOnePayDeviceOnLine(String deviceCode) {
        Date date = this.onePayJumpMap.get(deviceCode);
        if (date == null || DateUtils.addMinutes(date, this.jumpInterval).getTime() < System.currentTimeMillis()) {
            //主动请求一次
            if (this.onePayRequestActive(deviceCode)) {
                this.onePayJumpMap.put(deviceCode, new Date());
                return true;
            } else {
                log.error("机器掉线:" + deviceCode);
                return false;
            }
        }
        return true;
    }

    @Override
    public boolean checkOnePayDeviceOnLine(Integer id) {
        QueryWrapper<OzoneDevice> q1 = new QueryWrapper<>();
        q1.select("device_code");
        q1.eq("id", id);
        return this.checkOnePayDeviceOnLine(this.ozoneDeviceService.getOne(q1).getDeviceCode());
    }

    @Override
    public boolean checkMonthlyDeviceOnLine(String deviceCode) {
        Date date = this.monthlyJumpMap.get(deviceCode);
        if (date == null || DateUtils.addMinutes(date, this.jumpInterval).getTime() < System.currentTimeMillis()) {
            //主动请求一次
            if (this.monthlyRequestActive(deviceCode)) {
                this.monthlyJumpMap.put(deviceCode, new Date());
                return true;
            } else {
                log.error("机器掉线:" + deviceCode);
                return false;
            }
        }
        return true;
    }

    @Override
    public boolean checkMonthlyDeviceOnLine(Integer id) {
        QueryWrapper<MonthlyfeeDevice> q1 = new QueryWrapper<>();
        q1.select("device_code");
        q1.eq("id", id);
        MonthlyfeeDevice monthlyfeeDevice = this.monthlyfeeDeviceService.getOne(q1);
        if (monthlyfeeDevice == null) {
            return false;
        }
        return this.checkMonthlyDeviceOnLine(monthlyfeeDevice.getDeviceCode());
    }

    @Override
    public boolean checkBuyDeviceOnLine(String deviceCode) {
        Date date = this.buyJumpMap.get(deviceCode);
        if (date == null || DateUtils.addMinutes(date, this.jumpInterval).getTime() < System.currentTimeMillis()) {
            //主动请求一次
            if (this.buyRequestActive(deviceCode)) {
                this.buyJumpMap.put(deviceCode, new Date());
                return true;
            } else {
                log.error("机器掉线:" + deviceCode);
                return false;
            }
        }
        return true;
    }

    @Override
    public boolean checkBuyDeviceOnLine(Integer id) {
        return this.checkBuyDeviceOnLine("code");
    }

    /**
     * 主动请求,最后的查验
     */
    private boolean onePayRequestActive(String code) {
        this.onePayClient.pushMessage(code);
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Date date = this.onePayJumpMap.get(code);
        return date != null && DateUtils.addMinutes(date, this.jumpInterval).getTime() > System.currentTimeMillis();
    }

    private boolean monthlyRequestActive(String code) {
        this.monthlyClient.pushMessage(code);
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Date date = this.monthlyJumpMap.get(code);
        return date != null && DateUtils.addMinutes(date, this.jumpInterval).getTime() > System.currentTimeMillis();
    }

    private boolean buyRequestActive(String code) {
        this.buyClient.pushMessage(code);
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Date date = this.buyJumpMap.get(code);
        return date != null && DateUtils.addMinutes(date, this.jumpInterval).getTime() > System.currentTimeMillis();
    }

    private void refreshMap(Map<String, Date> map) {
        List<String> list = new ArrayList<>();
        long l = System.currentTimeMillis();
        for (Map.Entry<String, Date> entry : map.entrySet()) {
            if (DateUtils.addHours(entry.getValue(), 2).getTime() < l) {
                list.add(entry.getKey());
            }
        }
        for (String key : list) {
            map.remove(key);
        }
    }

    class OnePayCallback extends MyMqttCallback {

        @Override
        protected void connectionLost2(Throwable throwable) {

        }

        @Override
        protected void messageArrived2(String topic, MqttMessage mqttMessage) {
            String deviceCode = new String(mqttMessage.getPayload());
            JumpService_V2.this.receiveOnePayJump(deviceCode);
        }

        @Override
        protected void deliveryComplete2(IMqttDeliveryToken iMqttDeliveryToken) {

        }

    }

    class MonthlyCallback extends MyMqttCallback {

        @Override
        protected void connectionLost2(Throwable throwable) {

        }

        @Override
        protected void messageArrived2(String topic, MqttMessage mqttMessage) {
            String deviceCode = new String(mqttMessage.getPayload());
            JumpService_V2.this.receiveMonthlyJump(deviceCode);
        }

        @Override
        protected void deliveryComplete2(IMqttDeliveryToken iMqttDeliveryToken) {

        }

    }

    class BuyCallback extends MyMqttCallback {

        @Override
        protected void connectionLost2(Throwable throwable) {

        }

        @Override
        protected void messageArrived2(String topic, MqttMessage mqttMessage) {
            String deviceCode = new String(mqttMessage.getPayload());
            JumpService_V2.this.receiveBuyJump(deviceCode);
        }

        @Override
        protected void deliveryComplete2(IMqttDeliveryToken iMqttDeliveryToken) {

        }

    }

}
